package com.iqiyi.pps.epg.web.timer;

import com.iqiyi.pps.epg.core.model.channel.ChannelBase;
import com.iqiyi.pps.epg.core.model.channel.ChannelConfig;
import com.iqiyi.pps.epg.core.model.channel.ChannelLog;
import com.iqiyi.pps.epg.core.model.rtmp.RtmpServer;
import com.iqiyi.pps.epg.core.model.rtmp.RtmpServerStat;
import com.iqiyi.pps.epg.core.model.stream.LiveChannelStream;
import com.iqiyi.pps.epg.core.service.channel.ChannelService;
import com.iqiyi.pps.epg.core.service.program.IProgramAsyncService;
import com.iqiyi.pps.epg.core.service.rtmp.RtmpService;
import com.iqiyi.pps.epg.core.service.stream.LiveChannelStreamService;
import com.iqiyi.pps.epg.core.utils.*;
import org.apache.commons.lang3.StringUtils;
import org.dom4j.Document;
import org.dom4j.Element;
import org.dom4j.io.SAXReader;
import org.hibernate.Session;
import org.hibernate.SessionFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import java.io.BufferedInputStream;
import java.io.ByteArrayInputStream;
import java.sql.Timestamp;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * 所有服务器上的流信息同步、频道下码流检测
 * 每10分钟执行
 */
public class RtmpServerStatSyncTrigger {
    public final static int STATUS_IDLE = 1;
    public final static int STATUS_RUNNING = 2;
    private static Logger logger = LoggerFactory.getLogger(RtmpServerStatSyncTrigger.class);
    @Autowired
    private RtmpService rtmpService;
    @Autowired
    private ChannelService channelService;
    @Autowired
    private LiveChannelStreamService liveChannelStreamService;
    @Autowired
    private IProgramAsyncService programAsyncService;

    private Session session;
    private Map<String, RtmpServerStat> rtmpServerStatMap;
    //    private Map<String, RtmpServerStat> rtmpInfoMap;
    private Map<String, LiveChannelStream> liveChannelStreamMap;
    private Map<String, Integer> serverPortMap;
    private int acceptableInterval = 5 * 60 * 1000;
    private int status = 0;
    private int restRepeatTime = 0;
    private long startTaskTime = 0;

    public void task() {
        synchronized (this) {
            if (!needRunning()) {
                restRepeatTime++;
                logger.info("RtmpServerStatSyncTrigger is running! restRepeatTime={}: status={} ", restRepeatTime, status);
                return;
            }
            status = STATUS_RUNNING;
            restRepeatTime++;
        }


        while (restRepeatTime > 0) {
            logger.info("[RtmpServerStatSyncTrigger][restRepeatTime={}][status={}]", restRepeatTime, status);
            startTaskTime = System.currentTimeMillis();

            logger.info("[RtmpServerStatSyncTrigger][task][start][ThreadName={}][{}]", Thread.currentThread().getName(), new Date());
            try {
                SessionFactory sessionFactory = (SessionFactory) ApplicationContextUtil.getBean(SessionFactory.class);
                session = sessionFactory.openSession();

                // rtmp服务器列表
                List<RtmpServer> serverList = rtmpService.getServerList();

                if (null == serverList || serverList.size() == 0) {
                    logger.info("[RtmpServerStatSyncTrigger][serverList is empty]");
                    continue;
                }

                serverPortMap = transferServerList(serverList);

                // rtmp流列表
                List<RtmpServerStat> rtmpServerStatList = rtmpService.getServerStatList();
                if (rtmpServerStatList.size() > 0) {
                    rtmpServerStatMap = new HashMap<String, RtmpServerStat>();
//                rtmpInfoMap = new HashMap<>(rtmpServerStatList.size());
                    for (RtmpServerStat item : rtmpServerStatList) {
                        rtmpServerStatMap.put(item.getServerId() + item.getName(), item);
//                    rtmpInfoMap.put(item.getName(), item);
                    }
                } else {
                    rtmpServerStatMap = new HashMap<String, RtmpServerStat>();
//                rtmpInfoMap = new HashMap<>();
                }

                // 上线频道的流列表
                List<LiveChannelStream> liveChannelStreamList = liveChannelStreamService.getEnableChannelStream();
                liveChannelStreamMap = new HashMap(liveChannelStreamList.size());

                // (deprecated) getEnableChannelStream() 结果中已经过滤被删除的流.
                for (LiveChannelStream item : liveChannelStreamList) {
                    // 删除的不处理
                    if (item.getStatus() > LiveChannelStream.STATUS_DELETE) {
                        liveChannelStreamMap.put(item.getStreamName(), item);
                    }
                }

                logger.info("[RtmpServerStatSyncTrigger][task][part1][ThreadName={}][cost={}]", Thread.currentThread().getName(), System.currentTimeMillis() - startTaskTime);
                // 同步rtmp流
                ExecutorService exec = Executors.newFixedThreadPool(serverList.size());
                for (RtmpServer server : serverList) {
                    exec.execute(new TaskExecutor(server));
                }
                exec.shutdown();

                logger.info("[RtmpServerStatSyncTrigger][task][part2][ThreadName={}][cost={}]", Thread.currentThread().getName(), System.currentTimeMillis() - startTaskTime);

                // 多线程处理结束
                boolean errorOccure = false;
                try {
                    while (!exec.awaitTermination(10, TimeUnit.SECONDS)) {
                        if (System.currentTimeMillis() - startTaskTime > acceptableInterval) {
                            exec.shutdownNow();
                            errorOccure = true;
                            break;
                        }
                    }
                } catch (Exception e) {
                    logger.info("[RtmpServerStatSyncTrigger][task][awaitTermination][exception={}]", e);
                }

                if (errorOccure) {
                    logger.info("[RtmpServerStatSyncTrigger][task][task timeout, shutdown all tasks]");
                    continue;
                }

                logger.info("[RtmpServerStatSyncTrigger][task][part3][ThreadName={}][cost={}]", Thread.currentThread().getName(), System.currentTimeMillis() - startTaskTime);
                // 有未编码状态流时，隔10秒重新检测然后报警，下架频道
//            if (liveChannelStreamMap.size() > 0) {
//                logger.info("[RtmpServerStatSyncTrigger][task][retry][size={}]", liveChannelStreamMap.size());
//
//                Thread.sleep(10000); // 休息10秒
//
//                exec = Executors.newFixedThreadPool(serverList.size());
//                for (RtmpServer server : serverList) {
//                    exec.execute(new RetryExecutor(server));
//                }
//                exec.shutdown();
//
//                // 多线程处理结束
//                try {
//                    while (!exec.awaitTermination(10, TimeUnit.SECONDS));
//                } catch (Exception e) {
//                    logger.info("[RtmpServerStatSyncTrigger][task][awaitTermination][retry][exception={}]", e);
//                }
//
//                // 频道下线
//                logger.info("[RtmpServerStatSyncTrigger][task][retry][liveChannelStreamMap.size={}]", liveChannelStreamMap.size());
//                for (LiveChannelStream stream : liveChannelStreamMap.values()) {
//                    // 没找到的设置状态
//                    if (stream.getStatus() == LiveChannelStream.STATUS_ACTIVED) {
//                        stream.setStatus(LiveChannelStream.STATUS_IDLED);
//                        liveChannelStreamService.save(stream);
//                    }
//                    // 频道下线
//                    disableChannel(stream.getFatherId(), stream.getStreamName());
//                }
//            }

                // 设置直播流未编码
                Iterator<Map.Entry<String, LiveChannelStream>> it = liveChannelStreamMap.entrySet().iterator();
                while (it.hasNext()) {
                    Map.Entry<String, LiveChannelStream> entry = it.next();
                    // 没找到的设置状态
                    LiveChannelStream stream = entry.getValue();
                    int rtmpSync = (stream.getRtmpSync() == null ? 0 : stream.getRtmpSync()) + 1;
                    logger.info("[RtmpServerStatSyncTrigger][task][STATUS_IDLED][ThreadName={}][fatherId={}][streamName={}][rtmpSync={}]", Thread.currentThread().getName(), stream.getFatherId(), stream.getStreamName(), rtmpSync);
                    //连续3次未编码，才发邮件
                    if (rtmpSync < 3) {
                        stream.setRtmpSync(rtmpSync);
                        it.remove();
                    } else {
                        stream.setRtmpSync(0);
                    }

                    if (stream.getStatus() == LiveChannelStream.STATUS_ACTIVED) {
                        stream.setStatus(LiveChannelStream.STATUS_IDLED);
                        stream.setBitRate(0L);
                        stream.setInputBits("");
                        stream.setOutputBits("");
                        stream.setAspectRatio("");
                        stream.setFrameRate("");
                        stream.setScreenSize("");
                        stream.setVideo("");
                        stream.setAudio("");
                        LiveStreamManager.writeChannelLog(stream.getFatherId(), Constants.LogType.ENCODE_STOP.getIndex(),
                                "编码状态更新为未编码 [streamName=" + stream.getStreamName() + "]");
                        if (stream.getIsEnable()) {
                            programAsyncService.addTaskCheckProgramInLive(stream.getFatherId(), false, stream.getStreamName());
                        }
                    }

                    LiveChannelStream oldLiveChannelStream = liveChannelStreamService.getById(stream.getId());
                    if (null != oldLiveChannelStream) {
                        stream.setBackupEachOther(oldLiveChannelStream.getBackupEachOther());
                        stream.setSliceIndex(oldLiveChannelStream.getSliceIndex());
                        stream.setStreamUrls(oldLiveChannelStream.getStreamUrls());
                        stream.setDolbyStreamUrls(oldLiveChannelStream.getDolbyStreamUrls());
                        stream.setDolbyStreamUrl(oldLiveChannelStream.getDolbyStreamUrl());
                        liveChannelStreamService.save(stream);
                    }

                    // 停止未编码的流的flv,hcdn,ts和mp4
                    LiveStreamManager.stopAll(stream.getFatherId(), stream.getStreamName());
                    logger.info("[RtmpServerStatSyncTrigger][channelId={}][streamName={}][LiveStreamManager.stopAll]",
                            stream.getFatherId(), stream.getStreamName());
                }


                // 处理失效的流
                if (!rtmpServerStatMap.isEmpty()) {
                    logger.info("[RtmpServerStatSyncTrigger][task][rtmpServerStatSize={}]", rtmpServerStatMap.size());
                    for (RtmpServerStat item : rtmpServerStatMap.values()) {
                        if (item.getStatus() == RtmpServerStat.STATUS_VALID) {
                            item.setStatus(RtmpServerStat.STATUS_INVALID);
                            rtmpService.saveServerStat(item);
                        }
                    }
                }

                // 发送报警邮件
                if (liveChannelStreamMap.size() > 0) {
                    StringBuilder sb = new StringBuilder();
                    for (LiveChannelStream item : liveChannelStreamMap.values()) {
                        // 临时直播流不报警
                        if (item.getStreamName().startsWith("ls_")) {
                            continue;
                        }

                        ChannelBase channelBase = channelService.getChannelBase(item.getFatherId());

                        if (channelBase.getIsEnable() != ChannelBase.ENABLE || channelBase.getFormal() == 0) {
                            continue;
                        }

                        sb.append("频道ID：\t")
                                .append(item.getFatherId() + "<br />")
                                .append("奇谱ID：\t")
                                .append(channelBase.getQipuId() + "<br />")
                                .append("频道名称：\t")
                                .append(channelBase.getName() + "<br />")
                                .append("流名称：\t")
                                .append("<a href='http://")
                                .append(item.getStreamIp())
                                .append(":" + serverPortMap.get(item.getStreamIp()) + "/stat' target='_blank'>rtmp://")
                                .append(item.getStreamIp())
                                .append("/live/")
                                .append(item.getStreamName())
                                .append("</a><br />");
                    }
                    if (sb.length() == 0) {
                        continue;
                    }
                    List<ChannelConfig> mailList = channelService.getConfig(ChannelConfig.TEAM_TRIGGER_MAIL_LIST);
                    String[] tos = new String[mailList.size()];
                    int i = 0;
                    for (ChannelConfig item : mailList) {
                        tos[i] = item.getVal();
                        ++i;
                    }

                    SendMailUtils.send("直播流同步任务报警", tos, "以下流未编码状态<br /><br />" + sb.toString());
                }
                logger.info("[RtmpServerStatSyncTrigger][task][part4][ThreadName={}][cost={}]", Thread.currentThread().getName(), System.currentTimeMillis() - startTaskTime);
            } catch (Exception e) {
                logger.info("[RtmpServerStatSyncTrigger][task][exception=]", e);
            } finally {
                logger.info("[RtmpServerStatSyncTrigger][task][end][ThreadName={}][{}]", Thread.currentThread().getName(), new Date());
                restRepeatTime--;
            }
        }

        status = STATUS_IDLE;

    }

    public boolean needRunning() {
        return status != STATUS_RUNNING;
    }


    private void updateLiveChannelStream(RtmpServerStat rtmp, LiveChannelStream stream) {
        stream.setEncodedIp(rtmp.getAddress());
        stream.setInputBits(rtmp.getInputBits().toString());
        stream.setOutputBits(rtmp.getOutputBits().toString());
        stream.setFrameRate(rtmp.getFramerate().toString());
        stream.setBitRate(rtmp.getBitrate().longValue());
        stream.setVideo(rtmp.getVideo());
        stream.setAudio(rtmp.getAudio());

        int width = rtmp.getWidth(), height = rtmp.getHeight();
        stream.setScreenSize(width + ":" + height);
        stream.setAspectRatio("");
        if (width > 0 && height > 0) {
            // 画面比例
            double scale = Math.round(width * 100.0 / height) / 100.0;
            if (scale >= 1.5) {
                stream.setAspectRatio("16:9");
            } else if (scale <= 1.33) {
                stream.setAspectRatio("4:3");
            } else {
                stream.setAspectRatio("");
            }
        }
        if (rtmp.getActive() == RtmpServerStat.ACTIVE_VALID) {
            stream.setStatus(LiveChannelStream.STATUS_ACTIVED);
        } else {
            stream.setStatus(LiveChannelStream.STATUS_IDLED);
        }
    }

    private void disableChannel(long channelId, String streamName) {
        ChannelBase channelInfo = channelService.getChannelBase(channelId);
        String descr = streamName + " 流不在编码";
        // 常规直播才下线频道
        if (channelInfo.getLiveType() == ChannelBase.LIVE_TYPE_CT) {
            channelInfo.setIsEnable(ChannelBase.DISABLE);
            channelService.saveBase(channelInfo);
            descr += "，频道下线";
        }
        // 记日志
        ChannelLog log = new ChannelLog();
        log.setFatherId(channelInfo.getId());
        log.setType(203);
        log.setDescr(descr);
        log.setUser("_system_");
        channelService.saveLog(log);
    }

    private RtmpServerStat parseRtmpInfo(Object item, int serverId) {
        RtmpServerStat entity = new RtmpServerStat();
        entity.setServerId(serverId);
        entity.setActive(RtmpServerStat.ACTIVE_INVALID);
        entity.setStatus(RtmpServerStat.STATUS_VALID);
        // 以下有时候没有值
        entity.setAudio("");
        entity.setBitrate(0);
        entity.setFramerate(0);
        entity.setWidth(0);
        entity.setHeight(0);
        entity.setVideo("");
        entity.setAudio("");
        entity.setModifyTime(new Timestamp(System.currentTimeMillis()));

        try {
            Element node = (Element) item;
            Iterator<Element> iterator = node.elementIterator();
            while (iterator.hasNext()) {
                Element element = iterator.next();
                switch (element.getName().toLowerCase()) {
                    case "name":
                        entity.setName(element.getText());
                        break;
                    case "reconn":
                        entity.setReconn(Integer.valueOf(element.getText()));
                        break;
                    case "time":
                        entity.setUptime(Long.valueOf(element.getText()));
                        break;
                    case "in":
                        entity.setInBytes(Long.valueOf(element.getText()));
                        break;
                    case "out":
                        entity.setOutBytes(Long.valueOf(element.getText()));
                        break;
                    case "bwin":
                        entity.setInputBits(Long.valueOf(element.getText()));
                        break;
                    case "bwout":
                        entity.setOutputBits(Long.valueOf(element.getText()));
                        break;
                    case "client":
                        boolean isPublishing = false;
                        String address = "";
                        for (Iterator<Element> clientIterator = element.elementIterator(); clientIterator.hasNext(); ) {
                            Element clientElement = clientIterator.next();
                            switch (clientElement.getName().toLowerCase()) {
                                case "address":
                                    address = clientElement.getText();
                                    break;
                                case "publishing":
                                    isPublishing = true;
                                    break;
                            }
                        }

                        if (isPublishing) {
                            entity.setAddress(address);
                        }

                        break;
                    case "meta":
                        for (Iterator<Element> metaIterator = element.elementIterator(); metaIterator.hasNext(); ) {
                            Element metaElement = metaIterator.next();
                            switch (metaElement.getName().toLowerCase()) {
                                case "width":
                                    entity.setWidth(Integer.valueOf(metaElement.getText()));
                                    break;
                                case "height":
                                    entity.setHeight(Integer.valueOf(metaElement.getText()));
                                    break;
                                case "framerate":
                                    entity.setFramerate(Integer.valueOf(metaElement.getText()));
                                    break;
                                case "bitrate":
                                    entity.setBitrate(Integer.valueOf(metaElement.getText()));
                                    break;
                                case "video":
                                    entity.setVideo(metaElement.getText());
                                    break;
                                case "audio":
                                    entity.setAudio(metaElement.getText());
                                    break;
                            }
                        }
                        break;
                    case "active":
                        entity.setActive(RtmpServerStat.ACTIVE_VALID);
                        break;
                }
            }

            if (null == entity.getAddress()) {
                entity.setAddress(" ");
            }

            return entity;
        } catch (Exception e) {
            logger.error("[RtmpServerStatSyncTrigger][parseRtmpInfo][e={}]", e);
            return null;
        }
    }

    private RtmpServerStat parseNewRtmpInfo(Object item, int serverId) {
        RtmpServerStat entity = new RtmpServerStat();
        entity.setServerId(serverId);
        entity.setActive(RtmpServerStat.ACTIVE_INVALID);
        entity.setStatus(RtmpServerStat.STATUS_VALID);
        // 以下有时候没有值
        entity.setReconn(0);
        entity.setAudio("");
        entity.setBitrate(0);
        entity.setFramerate(0);
        entity.setWidth(0);
        entity.setHeight(0);
        entity.setVideo("");
        entity.setAudio("");
        entity.setModifyTime(new Timestamp(System.currentTimeMillis()));

        try {
            Element node = (Element) item;
            Iterator<Element> iterator = node.elementIterator();
            while (iterator.hasNext()) {
                Element element = iterator.next();
                switch (element.getName().toLowerCase()) {
                    case "name":
                        entity.setName(element.getText());
                        break;
                    case "time":
                        entity.setUptime(Long.valueOf(element.getText()));
                        break;
                    case "bytes_in":
                        entity.setInBytes(Long.valueOf(element.getText()));
                        break;
                    case "bytes_out":
                        entity.setOutBytes(Long.valueOf(element.getText()));
                        break;
                    case "bw_in":
                        entity.setInputBits(Long.valueOf(element.getText()));
                        break;
                    case "bw_out":
                        entity.setOutputBits(Long.valueOf(element.getText()));
                        break;
                    case "client":
                        boolean isPublishing = false;
                        String address = "";
                        for (Iterator<Element> clientIterator = element.elementIterator(); clientIterator.hasNext(); ) {
                            Element clientElement = clientIterator.next();
                            switch (clientElement.getName().toLowerCase()) {
                                case "address":
                                    address = clientElement.getText();
                                    break;
                                case "publishing":
                                    isPublishing = true;
                                    break;
                            }
                        }

                        if (isPublishing) {
                            entity.setAddress(address);
                        }

                        break;
                    case "meta":
                        for (Iterator<Element> metaIterator = element.elementIterator(); metaIterator.hasNext(); ) {
                            Element metaElement = metaIterator.next();
                            switch (metaElement.getName().toLowerCase()) {
                                case "video":
                                    for (Iterator<Element> videoIterator = metaElement.elementIterator(); videoIterator.hasNext(); ) {
                                        Element videoElement = videoIterator.next();
                                        switch (videoElement.getName().toLowerCase()) {
                                            case "width":
                                                entity.setWidth(Integer.valueOf(videoElement.getText()));
                                                break;
                                            case "height":
                                                entity.setHeight(Integer.valueOf(videoElement.getText()));
                                                break;
                                            case "frame_rate":
                                                entity.setFramerate(Integer.valueOf(videoElement.getText()));
                                                break;
                                            case "bit_rate":
                                                entity.setBitrate(Integer.valueOf(videoElement.getText()));
                                                break;
                                            case "codec":
                                                entity.setVideo(videoElement.getText());
                                                break;
                                        }
                                    }
                                    break;
                                case "audio":
                                    for (Iterator<Element> audioIterator = metaElement.elementIterator(); audioIterator.hasNext(); ) {
                                        Element audioElement = audioIterator.next();
                                        switch (audioElement.getName().toLowerCase()) {
                                            case "codec":
                                                entity.setAudio(audioElement.getText());
                                        }
                                    }
                                    break;
                            }
                        }
                        break;
                    case "active":
                        entity.setActive(RtmpServerStat.ACTIVE_VALID);
                        break;
                }
            }

            if (null == entity.getAddress()) {
                entity.setAddress(" ");
            }

            return entity;
        } catch (Exception e) {
            logger.error("[RtmpServerStatSyncTrigger][parseNewRtmpInfo][e={}]", e);
            return null;
        }
    }

    public static Map<String, Integer> transferServerList(List<RtmpServer> serverList) {
        Map<String, Integer> serverListMap = new HashMap<>();
        for (RtmpServer server : serverList) {
            serverListMap.put(server.getPublicIp(), server.getPort());
            serverListMap.put(server.getPrivateIp(), server.getPort());
        }

        return serverListMap;
    }

    private class TaskExecutor implements Runnable {
        private RtmpServer server;
        private boolean canDisable = false;

        public TaskExecutor(RtmpServer server) {
            this.server = server;
        }

        public TaskExecutor(RtmpServer server, boolean canDisable) {
            this.server = server;
            this.canDisable = canDisable;
        }

        @Override
        public void run() {
            try {
                String url = "http://" + server.getPrivateIp() + ":" + server.getPort() + "/stat";
                logger.info("[RtmpServerStatSyncTrigger][thread][url={}]", url);
                String urlContent = ApiHttpRequestUtil.originalGet(url, null, null);
                logger.info("[RtmpServerStatSyncTrigger][thread][url={}][urlContent={}]", url, urlContent);
                if (null == urlContent || "".equals(urlContent)) {
                    return;
                }
                SAXReader saxReader = new SAXReader();
                Document document = saxReader.read(new BufferedInputStream(new ByteArrayInputStream(urlContent.getBytes("utf-8"))));
                Element rootElement = document.getRootElement();
                List nodes = rootElement.selectNodes("//server/application/live/stream");
                if (nodes.size() > 0) {
                    for (Object item : nodes) {
                        try {
                            RtmpServerStat entity = server.getParseType() == 0 ? parseRtmpInfo(item, server.getId()) : parseNewRtmpInfo(item, server.getId());
                            if (null == entity) {
                                continue;
                            }

                            logger.info("[RtmpServerStatSyncTrigger][thread][getStatData][RtmpServerStat][streamName={}][serverIP={},{}][serverStatus={}]", entity.getName(), server.getPublicIp(), server.getPrivateIp(), entity.getActive());
                            // 是否已存在的流，rtmp流列表中没有的为新增
                            if (rtmpServerStatMap.containsKey(entity.getServerId() + entity.getName())) {
                                RtmpServerStat rtmpServerStat = rtmpServerStatMap.get(entity.getServerId() + entity.getName());
                                rtmpServerStat.copyData(entity);
                                rtmpService.saveServerStat(rtmpServerStat);
                                rtmpServerStatMap.remove(rtmpServerStat.getServerId() + rtmpServerStat.getName());
                            } else {
                                rtmpService.saveServerStat(entity);
                            }

                            // 检查上线频道的流编码状态、是否更新服务器、更新live_channel_stream表
                            if (liveChannelStreamMap.containsKey(entity.getName())) {
                                LiveChannelStream stream = liveChannelStreamMap.get(entity.getName());
                                //防止stream 重写旧数据,此时获取最新的stream信息
                                stream = liveChannelStreamService.getById(stream.getId());
//                            boolean streamUpdated = false;

                                if (entity.getActive() == RtmpServerStat.ACTIVE_VALID) {
                                    liveChannelStreamMap.remove(entity.getName()); // 优先移除

                                    // 判断是否更换了服务器
                                    if (!stream.getStreamIp().equals(server.getPublicIp())) {
                                        logger.info("[RtmpServerStatSyncTrigger][thread][changeServer][streamName={}][oldIp={}][newIp={}]", stream.getStreamName(), stream.getStreamIp(), server.getPublicIp());
                                        // rtmpChanged传的stream必须是老的数据
                                        String changedResult = LiveStreamManager.rtmpChanged(stream, server.getPublicIp());
                                        stream.setStreamIp(server.getPublicIp());

                                        logger.info("[RtmpServerStatSyncTrigger][thread][LiveStreamManager.rtmpChanged={}]", changedResult);
                                    }

                                    // 上线频道的流未启动的要启动各播放格式的流，此时库里码流需要是已编码状态，所以下面update流信息
                                    if (stream.getStatus() == LiveChannelStream.STATUS_IDLED) {
                                        logger.info("[RtmpServerStatSyncTrigger][thread][startAuto][id={}][streamName={}][status={}]", stream.getId(), stream.getStreamName(), stream.getStatus());
                                        // update 流信息 auto 方法需要最新的数据
                                        updateLiveChannelStream(entity, stream);
                                        if (stream.getIsEnable()) {
                                            programAsyncService.addTaskCheckProgramInLive(stream.getFatherId(), true, stream.getStreamName());
                                        }
                                        liveChannelStreamService.save(stream);
//                                    streamUpdated = true;

                                        String autoResult = LiveStreamManager.auto(stream.getFatherId(), entity.getName(), server.getPublicIp(), stream.getIsEnable(), stream.getStreamApiVersion(),
                                                StringUtils.isNotEmpty(stream.getDolbyStreamUrl()));
                                        logger.info("[RtmpServerStatSyncTrigger][thread][channelId={}][streamName={}][serverId={}][LiveStreamManager.auto={}]", stream.getFatherId(), entity.getName(), server.getPublicIp(), autoResult);
                                    } else {
                                        updateLiveChannelStream(entity, stream);
                                        liveChannelStreamService.save(stream);
                                    }
                                    logger.info("[RtmpServerStatSyncTrigger][thread][LiveChannelStream][Update][fatherId={}][streamName={}][streamIp={}][streamStatus={}]", stream.getFatherId(), stream.getStreamName(), stream.getStreamIp(), stream.getStatus());

                                } else {
                                    // 流不在编码时更新频道启用状态为无效
                                    if (canDisable) {
                                        disableChannel(stream.getFatherId(), entity.getName());
                                    }
                                }
                            }
                        } catch (Exception e) {
                            logger.error("[RtmpServerStatSyncTrigger][thread][exception={}]", e);

                            RtmpServerStat entity = server.getParseType() == 0 ? parseRtmpInfo(item, server.getId()) : parseNewRtmpInfo(item, server.getId());

                            if (null != entity) {

                                //RtmpServer获取数据异常，移除该RtmpServer下的直播流
                                if (liveChannelStreamMap.containsKey(entity.getName())) {
                                    LiveChannelStream stream = liveChannelStreamMap.get(entity.getName());
                                    String streamName = stream.getStreamName();
                                    logger.info("[RtmpServerStatSyncTrigger][thread][exception][remove][LiveChannelStream][channelId={}][streamName={}]", stream.getFatherId(), streamName);
                                    liveChannelStreamMap.remove(streamName);

                                    // Insert error log into the table of 'channel_log'
                                    ChannelLog channelLog = new ChannelLog();
                                    channelLog.setFatherId(stream.getFatherId());
                                    channelLog.setType(105);
                                    channelLog.setUser("_system_");
                                    channelLog.setDescr("定时器同步频道流信息错误!");

                                    channelService.saveLog(channelLog);

                                }

                                String streamName = entity.getName();
                                Integer serverId = entity.getServerId();
                                if (rtmpServerStatMap.containsKey(serverId + streamName)) {
                                    logger.info("[RtmpServerStatSyncTrigger][thread][exception][remove][RtmpServerStat][serverId={}][streamName={}]", serverId, streamName);
                                    rtmpServerStatMap.remove(serverId + streamName);
                                }
                            }

                        }
                    }
                }
            } catch (Exception e) {
                logger.error("[RtmpServerStatSyncTrigger][thread][privateIp={}][exception={}]", server.getPrivateIp(), e);

                List<LiveChannelStream> liveChannelStreams = liveChannelStreamService.getByStreamIp(server.getPublicIp());
                //RtmpServer获取数据异常，移除该RtmpServer下的直播流
                for (LiveChannelStream liveChannelStream : liveChannelStreams) {
                    String streamName = liveChannelStream.getStreamName();
                    if (liveChannelStreamMap.containsKey(streamName)) {
                        logger.info("[RtmpServerStatSyncTrigger][thread][exception][remove][LiveChannelStream][streamName={}]", streamName);
                        liveChannelStreamMap.remove(streamName);
                    }
                }

                List<RtmpServerStat> rtmpServerStats = rtmpService.getServerStatListByServerId(server.getId());
                for (RtmpServerStat rtmpServerStat : rtmpServerStats) {
                    String streamName = rtmpServerStat.getName();
                    Integer serverId = rtmpServerStat.getServerId();
                    if (rtmpServerStatMap.containsKey(serverId + streamName)) {
                        logger.info("[RtmpServerStatSyncTrigger][thread][exception][remove][RtmpServerStat][serverId={}][streamName={}]", serverId, streamName);
                        rtmpServerStatMap.remove(serverId + streamName);
                    }
                }
            }
        }
    }

//    private class RetryExecutor implements Runnable {
//        private RtmpServer server;
//
//        public RetryExecutor(RtmpServer server) {
//            this.server = server;
//        }
//
//        public void run() {
//            try {
//                String url = "http://" + server.getPrivateIp() + ":8080/stat";
//                logger.info("[RtmpServerStatSyncTrigger][retry][thread][url={}]", url);
//                SAXReader saxReader = new SAXReader();
//                Document document = saxReader.read(url);
//                Element rootElement = document.getRootElement();
//                List nodes = rootElement.selectNodes("//server/application/live/stream");
//                if (nodes.size() > 0) {
//                    String changedResult, autoResult;
//                    for (Object item : nodes) {
//                        RtmpServerStat entity = parseRtmpInfo(item, server.getId());
//
//                        // 重试只检查上次未检测到的流
//                        if (!liveChannelStreamMap.containsKey(entity.getName())) {
//                            continue;
//                        }
//
//                        LiveChannelStream stream = liveChannelStreamMap.get(entity.getName());
//                        boolean streamUpdated = false;
//
//                        // 在编码中
//                        if (entity.getActive() == RtmpServerStat.ACTIVE_VALID) {
//                            // 判断是否更换了服务器
//                            if (!stream.getStreamIp().equals(server.getPublicIp())) {
//                                logger.info("[RtmpServerStatSyncTrigger][retry][thread][changeServer][id={}][oldIp={}][newIp={}]", stream.getId(), stream.getStreamIp(), server.getPublicIp());
//                                // rtmpChanged传的stream必须是老的数据
//                                changedResult = LiveStreamManager.rtmpChanged(stream, server.getPublicIp());
//                                logger.info("[RtmpServerStatSyncTrigger][retry][thread][LiveStreamManager.rtmpChanged={}]", changedResult);
//                            }
//
//                            // 上线频道的流未启动的要启动各播放格式的流，此时库里码流需要是已编码状态，所以下面update流信息
//                            if (stream.getStatus() == LiveChannelStream.STATUS_IDLED) {
//                                logger.info("[RtmpServerStatSyncTrigger][retry][thread][startRtmp][id={}][streamName={}]", stream.getId(), stream.getStreamName());
//                                // update 流信息 auto 方法需要最新的数据
//                                updateLiveChannelStream(entity, stream);
//                                stream.setStreamIp(server.getPublicIp());
//                                liveChannelStreamService.save(stream);
//                                streamUpdated = true;
//
//                                autoResult = LiveStreamManager.auto(stream.getFatherId(), entity.getName(), server.getPublicIp());
//                                logger.info("[RtmpServerStatSyncTrigger][retry][thread][LiveStreamManager.auto={}]", autoResult);
//                            }
//
//                            liveChannelStreamMap.remove(entity.getName());
//                            logger.info("[RtmpServerStatSyncTrigger][retry][thread][actived][name={}]", entity.getName());
//                        } else {
//                            // 不在编码中
//                        }
//
//                        if (!streamUpdated) {
//                            updateLiveChannelStream(entity, stream);
//                            stream.setStreamIp(server.getPublicIp());
//                            liveChannelStreamService.save(stream);
//                        }
//
//                        RtmpServerStat temp = rtmpInfoMap.get(entity.getName());
//                        // 临时直播流是可以直接添加的
//                        if (temp != null) {
//                            session.evict(temp);
//                            rtmpServerStatMap.remove(entity.getName());
//                            entity.setId(temp.getId());
//                        }
//                        rtmpService.saveServerStat(entity);
//                    }
//                }
//            } catch (Exception e) {
//                logger.info("[RtmpServerStatSyncTrigger][retry][thread][exception=]", e);
//            }
//        }
//    }
}
